#include "ElasticsearchClient.h"
#include "util.h"
#include <HTTPLib/httplib.h>
#include <rapidjson/document.h>
#include <utility>

using namespace std;
using namespace httplib;
using namespace rapidjson;

ElasticsearchClient::ElasticsearchClient(string ip, int port) : ip(std::move(ip)), port(port) { }

ElasticsearchClient::~ElasticsearchClient() = default;

string ElasticsearchClient::GetErrorString()
{
    std::string str;
    str.swap(errorString);
    return str;
}

bool ElasticsearchClient::Ok()
{
    Client client(ip, port);
    auto resp = client.Get("/");
    if (!resp)
        return false;
    if (resp->status != 200)
        return false;
    return true;
}

Status ElasticsearchClient::CreateIndex(const string &indexName)
{
    auto path = stringstream();
    path << "/" << indexName;
    auto body = R"({"mappings":{"properties":{"row_key":{"type":"keyword"},"column":{"type":"keyword"},"value_k":{"type":"keyword"},"value_t":{"type":"text","analyzer":"ik_max_word","search_analyzer":"ik_smart","search_quote_analyzer":"ik_smart"}}}})";
    Client client(ip, port);
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    auto resp = client.Put(path.str().c_str(), headers, body, "application/json");
    if (!resp)
        return Failed;
    if (resp->status != 200)
    {
        errorString = resp->body;
        return Failed;
    }
    return Successful;
}

Status ElasticsearchClient::DeleteIndex(const std::string &indexName)
{
    auto path = stringstream();
    path << "/" << indexName;
    Client client(ip, port);
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    auto resp = client.Delete(path.str().c_str(), headers);
    if (!resp)
        return Failed;
    if (resp->status == 200)
        return Successful;

    auto dom = Document();
    dom.Parse(resp->body.c_str());
    if (dom.HasParseError())
        return Failed;
    if (!dom.HasMember("status") || !dom["status"].IsNumber())
        return Failed;
    if (dom["status"].GetUint64() == 404)
    {
        errorString = "Index not be found.";
        return Nonfatal;
    }
    return Failed;
}

Status ElasticsearchClient::CreateDocument(const string &indexName, const string &rowKey, const string &column,
                                           const string &value, bool canDivide)
{
    auto path = stringstream();
    path << "/" << indexName << "/_doc/" << Base64Encode(column) << "@" << Base64Encode(rowKey);
    auto body = stringstream();
    body << R"({"row_key":")" << rowKey << R"(","column":")" << column << R"(",)"
        << (canDivide ? R"("value_t":")" : R"("value_k":")") << value << R"("})";
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Put(path.str().c_str(), headers, body.str(), "application/json");
    if (!resp)
        return Failed;
    if (resp->status != 201)
        return Failed;
    return Successful;
}

Status ElasticsearchClient::CreateDocument(const string &indexName, const string &rowKey,
                                           std::list<std::tuple<std::string, std::string, bool>> cells)
{
    auto path = "/_bulk";
    stringstream body;
    for (auto &i: cells)
    {
        body << R"({"index":{"_index":")";
        body << indexName;
        body << R"(","_type":"_doc","_id":")";
        body << Base64Encode(get<0>(i));
        body << "@";
        body << Base64Encode(rowKey);
        body << R"("}})";
        body << "\n";
        body << R"({"row_key":")";
        body << rowKey;
        body << R"(","column":")";
        body << get<0>(i);
        body << R"(",)";
        body << (get<2>(i) ? R"("value_t":")" : R"("value_k":")");
        body << get<1>(i);
        body << R"("})";
        body << "\n";
    }
    Client client(ip, port);
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    auto resp = client.Put(path, headers, body.str(), "application/json");
    if (!resp)
        return Failed;
    if (resp->status != 200)
        return Failed;
    return Successful;
}

Status ElasticsearchClient::CreateDocument(const string &indexName, const ESCells &rows)
{
    auto path = "/_bulk";
    auto body = rows.ToString(indexName);
    Client client(ip, port);
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    auto resp = client.Put(path, headers, body, "application/json");
    if (!resp)
        return Failed;
    if (resp->status != 200)
        return Failed;
    return Successful;
}

Status ElasticsearchClient::DeleteDocument(const string &indexName, const string &rowKey, const string &column)
{
    auto path = stringstream();
    path << "/" << indexName << "/_doc/" << Base64Encode(column) << "@" << Base64Encode(rowKey);
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Delete(path.str().c_str(), headers);
    if (!resp)
        return Failed;
    if (resp->status == 200)
        return Successful;
    auto dom = Document();
    dom.Parse(resp->body.c_str());
    if (dom.HasParseError())
        return Failed;
    if (!dom.HasMember("result") || !dom["result"].IsString())
        return Failed;
    if (strcmp(dom["result"].GetString(), "not_found") == 0)
    {
        errorString = "Cell not be found.";
        return Nonfatal;
    }
    return Failed;

}

Status ElasticsearchClient::DeleteDocument(const string &indexName, const string &rowKey)
{
    auto path = stringstream();
    path << "/" << indexName << "/_delete_by_query";
    auto body = stringstream();
    body << R"({"query":{"match":{"row_key":")" << rowKey << R"("}}})";
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Post(path.str().c_str(), headers, body.str(), "application/json");
    if (!resp)
        return Failed;
    if (resp->status != 200)
        return Failed;
    return Successful;
}

optional<list<tuple<string, string, string>>>
ElasticsearchClient::SearchInIndex(const string &indexName, const string &keywords, size_t from, size_t size)
{
    auto path = stringstream();
    path << "/" << indexName << "/_search";
    auto body = stringstream();
    body << R"({"query":{"multi_match":{"query":")" << ReplaceChar(keywords, '\"', '\'')
        << R"(","fields": ["value_k","value_t"]}})"
        << (size == 0 ? "}" : (R"(,"from":)" + to_string(from) + R"(,"size":)" + to_string(size) + "}"));
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client(ip, port);
    auto resp = client.Get(path.str().c_str(), headers, body.str(), "application/json");
    if (!resp)
        return nullopt;
    if (resp->status != 200)
        return nullopt;
    auto result = list<tuple<string, string, string>>();
    Document dom;
    dom.Parse(resp->body.c_str());
    auto tab = (dom["hits"].GetObject())["hits"].GetArray();
    for (auto i = tab.Begin(); i != tab.End(); i++)
    {
        auto obj = (*i).GetObject()["_source"].GetObject();
        auto rowKey = obj["row_key"].GetString();
        auto column = obj["column"].GetString();
        auto value = obj.HasMember("value_t") ? obj["value_t"].GetString() : obj["value_k"].GetString();
        result.emplace_back(tuple{rowKey, column, value});
    }
    return result;
}



